//==========================================================================
// Copyright (c) 2000-2008,  Elastos, Inc.  All Rights Reserved.
//==========================================================================

#include <compoundChannel.h>
#include <osdepend.h>
#include "ChannelCommand.h"
#include "PkgManager.h"

#define MAX_TRANS_DATA_SIZE 0x800
#define MAX_STORAGE_SIZE 4092 // 4k - 4, kmalloc.h c_uMaxAllocationSize

CommunicationObject *FindObjByChannelNumber(list *pList, C_UINT channelNumber,
    C_LOCK lock)
{
    assert(NULL != pList);
    assert(0 != channelNumber);

    CommunicationObject *pObj = NULL;
    list *pNode;
    Lock(lock);
    pNode = NextNode(pList);
    while (pNode != pList) {
        pObj = GET_LIST_OBJ(pNode, CommunicationObject, m_entry);
        if (channelNumber == pObj->m_channelNumber) {
            UnLock(lock);
            return pObj;
        }
        pNode = NextNode(pNode);
    }
    UnLock(lock);
    return NULL;
}

C_INT DoChannelCommand(C_VOID *lpParam)
{
    ChannelCommand *cc = (ChannelCommand *) lpParam;

    CommunicationObject *pObj = NULL;
    switch (cc->GetCommand()) {
        case ChannelCommand::TYPE_CREATE:
            pObj = FindObjByChannelNumber(&CommunicationObject::m_sRunningList,
                    cc->GetChannel(),
                    CommunicationObject::m_sRunnListLock);
            if (NULL != pObj) {//Have been created
                pObj->SendCreateOkReply();
            }
            else {
                CommunicationObject::CreateCommuObj((PROPERTY)cc->GetProperty(),
                        cc->GetChannel() << 16, NULL);
            }
            break;
        case ChannelCommand::TYPE_REPLY:
            pObj = FindObjByChannelNumber(&CommunicationObject::m_sInitList,
                    cc->GetChannel(),
                    CommunicationObject::m_sInitListLock);
            if (NULL != pObj) {
                EventSet(pObj->m_event);
            }
            else { // If not found, notify that the channel is deleted!
                ChannelCommand ccDel(cc->GetChannel(), ChannelCommand::TYPE_DELETE);
                SendChannelData(&ccDel);
            }
            break;
        case ChannelCommand::TYPE_DELETE:
            CommunicationObject::DeleCommuObj(cc->GetChannel(), TRUE);
            break;
        case ChannelCommand::TYPE_RESET:
            ResetCompoundChannel();
            break;
        default:
            COMMU_PRINTF("[Err]: Wrong channel command.\n");
            break;
    }
    delete cc;
    return 0;
}

C_INT PutDataInStorageBuffer(CommunicationObject *obj, C_VOID *pBuffer,
    C_UINT size)
{
    C_UINT storageSize = 0;
    C_CHAR *pStorage = NULL;
    Lock(obj->m_lock);
    storageSize = obj->m_receiveStorageSize + size;

    pStorage = (C_CHAR *)MALLOC(storageSize);
    if (NULL == pStorage) {
        COMMU_PRINTF("[Err]: PutDataInStorageBuffer failed! file%s. line:%d.\n",
            __FILE__, __LINE__);
        UnLock(obj->m_lock);
        return C_OUT_OF_MEMORY;
    }
    memset(pStorage, 0, storageSize);
    // Storage already has data.
    if (0 != obj->m_receiveStorageSize) {
        memcpy(pStorage, obj->m_pReceiveStorageAddr, obj->m_receiveStorageSize);
        FREE(obj->m_pReceiveStorageAddr);
        obj->m_pReceiveStorageAddr = NULL;
    }

    memcpy(pStorage + obj->m_receiveStorageSize, (C_CHAR *)pBuffer, size);

    obj->m_receiveStorageSize = storageSize;
    obj->m_pReceiveStorageAddr = pStorage;
    UnLock(obj->m_lock);

    return 0;
}

/*
 * Check the object's storage buffer and wait util buffer's size is not too
 * large.
 */
C_VOID CheckStorageBufferSize(CommunicationObject *obj, C_UINT size)
{
    if (obj->m_receiveStorageSize + size > MAX_STORAGE_SIZE) { // 8k
    	C_EVENT evt = EventCreate(FALSE); // local event is created for sleep
	    while (obj->m_receiveStorageSize + size > MAX_STORAGE_SIZE) {
	    	WaitForEvent(evt, 50); // sleep for 50ms, 8k * 1000 / 50 = 160k/s
	    }
	    DeleEvent(evt);
	}
}

C_INT PutDataInReceiveBuffer(CommunicationObject *obj, C_VOID *pBuffer,
    C_UINT size)
{
    assert(obj);
    assert(pBuffer);
    assert(size);

    C_UINT theMoreDataSize = 0;
    C_INT ret = 0;

    bool bBufferReady = false;
    Lock(obj->m_lock);
    bBufferReady = (0 != obj->m_receiveSize) && (NULL != obj->m_pReceiveAddr);
    // No one to read, send to storage
    if (!bBufferReady || 0 != obj->m_receiveStorageSize) {
    	if (obj->m_receiveStorageSize + size > MAX_STORAGE_SIZE) {
			UnLock(obj->m_lock);
			CheckStorageBufferSize(obj, size);
			Lock(obj->m_lock);
		    bBufferReady = (0 != obj->m_receiveSize) && (NULL != obj->m_pReceiveAddr);
			// Check again
			if (!bBufferReady || 0 != obj->m_receiveStorageSize) {
				ret = PutDataInStorageBuffer(obj, pBuffer, size);
				UnLock(obj->m_lock);
				return ret;
			}
    	} else {
			ret = PutDataInStorageBuffer(obj, pBuffer, size);
			UnLock(obj->m_lock);
			return ret;
    	}
    }

    // If no storage data ,we put receive data into the receive address

    //Some one wait to read (no storage data to read).
    assert(obj->m_pReceiveAddr);
    assert(obj->m_receiveSize);

    // Recevice part of data, not enough
    if (size <= obj->m_receiveSize) {
        memcpy(obj->m_pReceiveAddr + obj->m_receiveOffset, pBuffer, size);
        obj->m_receiveOffset += size;
        obj->m_receiveSize -= size;
        UnLock(obj->m_lock);
        if (!EventSet(obj->m_event)) { // early notification
            COMMU_PRINTF("[ERR]: SetEvent failed! file%s. line:%d. ChannelNumber:%d.\n",
                    __FILE__, __LINE__, obj->m_channelNumber);
        }
        return 0;
    }

    //Recevice data large enough or more
    theMoreDataSize = size - obj->m_receiveSize;
    memcpy(obj->m_pReceiveAddr + obj->m_receiveOffset, pBuffer, obj->m_receiveSize);
    obj->m_receiveOffset += obj->m_receiveSize;
    ret = PutDataInStorageBuffer(obj,
            (C_VOID *)((C_CHAR *)pBuffer + obj->m_receiveSize),
            theMoreDataSize);
    //Clean the info to receive the next
    obj->m_receiveSize = 0; //Mission complete clean the signal data
    UnLock(obj->m_lock);

    if (!EventSet(obj->m_event)) {
        COMMU_PRINTF("ERR: SetEvent failed! file%s. line:%d."
                        " ChannelNumber:%d.\n",
                        __FILE__, __LINE__, obj->m_channelNumber);
    }

    return ret;
}

C_INT ReceiveLoop(C_VOID *arg)
{
    CommunicationObject *pObj = NULL;
    int err = 0;

    while (1) {
        err = 0;
        ChannelData *cd = ReceiveChannelData(&err);
        if (err != 0) {
            return -1;
        }
        if (NULL == cd) {
            COMMU_PRINTF("ERR: No package.\n");
        }
        else {
            if (0 == cd->GetChannelNumber()) {
                // channel command on channel#0
                CreateChannelThread(DoChannelCommand, cd, NULL);
                // variable cd will be released in DoReceiveThread method
            }
            else {
                // channel data, try to store data into buffer
                pObj = FindObjByChannelNumber(&CommunicationObject::m_sRunningList,
                        cd->GetChannelNumber(),
                        CommunicationObject::m_sRunnListLock);
                if (NULL != pObj) {
                    PutDataInReceiveBuffer(pObj, cd->GetData(), cd->GetDataSize());
                }
                else {
                    pObj = FindObjByChannelNumber(&CommunicationObject::m_sInitList,
                            cd->GetChannelNumber(),
                            CommunicationObject::m_sInitListLock);
                    if (NULL != pObj) {
                    	CheckStorageBufferSize(pObj, cd->GetDataSize());
                        PutDataInStorageBuffer(pObj, cd->GetData(), cd->GetDataSize());
                    }
                }
                delete cd;
            }
        }
    }
}
